package vip.xiaonuo.common.sses.util;

import cn.hutool.script.ScriptUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import vip.xiaonuo.common.cache.CommonRedisUtil;
import vip.xiaonuo.common.sses.entity.RedisEvent;
import vip.xiaonuo.common.sses.entity.SseMessage;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.*;

/**
 * @author zjt
 * @description
 * @date 2025/4/29 13:45
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class CommonSseUtil implements InitializingBean, DisposableBean {

	private final CommonRedisUtil redisUtil;

	private ExecutorService virtualThreadExecutor;

	private ScheduledExecutorService virtualThreadScheduler;

	private final Map<String, SseEmitter> localSseMap = new ConcurrentHashMap<>();
	private final Map<String, String> localLoginIdClientIdMap = new ConcurrentHashMap<>();

	private static final long SSE_CONNECTION_EXPIRE_SECONDS = 30;
	private static final String SSE_CONNECTION_CLIENT_KEY_PREFIX = "sse:connection:client_";
	private static final String SSE_CONNECTION_USER_KEY_PREFIX = "sse:connection:user_";
	private static final String SSE_MESSAGE_USER = "sse:message";

	@Override
	public void afterPropertiesSet() throws Exception {
		// 初始化虚拟线程执行器
		virtualThreadExecutor = Executors.newVirtualThreadPerTaskExecutor();
		// 初始化调度器
		virtualThreadScheduler = Executors.newScheduledThreadPool(1, Thread.ofVirtual().factory());
		virtualThreadScheduler.scheduleAtFixedRate(() -> virtualThreadExecutor.execute(this::heartbeatCheck), 0, 15,
				TimeUnit.SECONDS);
		// 订阅Redis消息
		redisUtil.subscribe(SSE_MESSAGE_USER, (message, pattern) -> virtualThreadExecutor
				.execute(() -> handleRedisMessage(new String(message.getBody()))));
	}

	private void shutdownExecutor(ExecutorService executor) {
		if (executor != null) {
			executor.shutdown(); // 拒绝新任务并执行队列中剩余任务
			try {
				if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
					executor.shutdownNow(); // 强制取消未完成任务
				}
			} catch (InterruptedException e) {
				executor.shutdownNow(); // 中断时立即终止
				Thread.currentThread().interrupt(); // 恢复中断状态
			}
		}
	}

	@Override
	public void destroy() throws Exception {
		shutdownExecutor(virtualThreadScheduler);
		shutdownExecutor(virtualThreadExecutor);
	}

	// 创建连接
	public SseEmitter createConnection(String clientId, String loginId) {
		// 双重检查锁模式
		SseEmitter emitter = localSseMap.get(clientId);
		if (emitter != null) {
			return emitter;
		}
		// 创建新连接
		emitter = new SseEmitter(0L);
		emitter.onCompletion(() -> removeConnection(clientId));
		emitter.onTimeout(() -> removeConnection(clientId));
		emitter.onError(ex -> removeConnection(clientId));
		localSseMap.put(clientId, emitter);
		localLoginIdClientIdMap.put(loginId, clientId);
		redisUtil.setCacheStringWithTimeOut(SSE_CONNECTION_CLIENT_KEY_PREFIX + clientId, loginId,
				SSE_CONNECTION_EXPIRE_SECONDS);
		redisUtil.setCacheStringWithTimeOut(SSE_CONNECTION_USER_KEY_PREFIX + loginId, clientId,
				SSE_CONNECTION_EXPIRE_SECONDS);
		return emitter;
	}

	/**
	 * 发送消息到指定客户端(通过Redis广播)
	 *
	 * @param loginId
	 * @param sseMessage
	 */
	public void sendMessageToUser(String loginId, SseMessage sseMessage) {
		// 直接发布到Redis通道
		redisUtil.publish(SSE_MESSAGE_USER, serialize(new RedisEvent(loginId, sseMessage)),
				s -> log.info("SSE：发布成功,{}", s));
	}

	public void sendMessageToAll(SseMessage message) {
		List<Object> clientIdList = redisUtil.getAllValuesByPrefix(SSE_CONNECTION_USER_KEY_PREFIX);
		for (Object clientId : clientIdList) {
			// 获取LoginId
			Optional<String> loginId = localLoginIdClientIdMap.entrySet().stream()
					.filter(entry -> entry.getValue().equals(clientId)).map(Map.Entry::getKey).findAny();
			loginId.ifPresentOrElse(
					// 当loginId存在时的处理逻辑
					id -> sendMessageToUser(id, message), () -> {
						// 当loginId不存在时的处理逻辑
						String redisVal = (String) redisUtil.get(SSE_CONNECTION_CLIENT_KEY_PREFIX + clientId);
						if (StringUtils.isNotBlank(redisVal)) {
							sendMessageToUser(redisVal, message);
						}
					});
		}
	}

	public void removeConnection(String clientId) {
		removeLocalConnection(clientId);
		redisUtil.remove(SSE_CONNECTION_CLIENT_KEY_PREFIX + clientId);
	}

	private void removeLocalConnection(String clientId) {
		Optional<String> loginId = localLoginIdClientIdMap.entrySet().stream()
				.filter(entry -> entry.getValue().equals(clientId)).map(Map.Entry::getKey).findAny();
		removeSseMap(clientId);
		loginId.ifPresent(this::removeLoginMap);
	}

	private void removeSseMap(String clientId) {
		SseEmitter emitter = localSseMap.remove(clientId);
		if (emitter != null) {
			try {
				emitter.complete();
			} catch (Exception e) {
				log.debug("SSE连接关闭异常", e);
			}
		}
	}

	private void removeLoginMap(String loginId) {
		localLoginIdClientIdMap.remove(loginId);
		redisUtil.remove(SSE_CONNECTION_USER_KEY_PREFIX + loginId);
	}

	private void heartbeatCheck() {
		try {
			localSseMap.keySet().forEach(clientId -> virtualThreadExecutor.execute(() -> {
				try {
					// 1. 发送心跳包检测连接活性
					localSseMap.get(clientId).send(SseEmitter.event().name("heartbeat").data("ping")
							.reconnectTime(SSE_CONNECTION_EXPIRE_SECONDS * 3000));
					// 2. 仅当心跳发送成功后才续期Redis
					// 续费LoginId
					Optional<String> loginId = localLoginIdClientIdMap.entrySet().stream()
							.filter(entry -> entry.getValue().equals(clientId)).map(Map.Entry::getKey).findAny();
					loginId.ifPresent(s -> {
						redisUtil.setCacheStringWithTimeOut(SSE_CONNECTION_CLIENT_KEY_PREFIX + clientId, s,
								SSE_CONNECTION_EXPIRE_SECONDS);
						redisUtil.setCacheStringWithTimeOut(SSE_CONNECTION_USER_KEY_PREFIX + s, clientId,
								SSE_CONNECTION_EXPIRE_SECONDS);
					});
					log.debug("心跳检测成功并续期连接: {}", clientId);
				} catch (IOException e) {
					// 3. 心跳失败处理
					log.warn("心跳检测失败，移除无效连接: {}", clientId);
					removeConnection(clientId);
				} catch (Exception e) {
					log.error("连接续期异常: {}", clientId, e);
				}
			}));
		} catch (Exception e) {
			log.error("连接续期任务异常", e);
		}
		log.debug("SSE：当前存活的ClientId:{}", localSseMap.keySet());
	}

	/**
	 * 订阅消息
	 */
	private void handleRedisMessage(String messageJson) {
		try {
			RedisEvent message = deserialize(messageJson);
			String loginId = message.getLoginId();
			if (localLoginIdClientIdMap.containsKey(loginId)) {
				String clientId = localLoginIdClientIdMap.get(loginId);
				if (localSseMap.containsKey(clientId)) {
					SseEmitter emitter = localSseMap.get(clientId);
					try {
						log.info("handleRedisMessage loginId:{}", message.getLoginId());
						emitter.send(SseEmitter.event().data(message.getData()));
						// 发送成功后续期连接
						redisUtil.setExpireIfKeyExists(SSE_CONNECTION_CLIENT_KEY_PREFIX + clientId,
								SSE_CONNECTION_EXPIRE_SECONDS);
						redisUtil.setExpireIfKeyExists(SSE_CONNECTION_USER_KEY_PREFIX + loginId,
								SSE_CONNECTION_EXPIRE_SECONDS);
					} catch (IOException e) {
						removeConnection(clientId);
					}
				}
			}
		} catch (Exception e) {
			log.error("处理SSE消息异常", e);
		}
	}

	public static String serialize(RedisEvent redisEvent) {
		return JSON.toJSONString(redisEvent);
	}

	public static RedisEvent deserialize(String json) {
		return JSONObject.parseObject((String) ScriptUtil.eval(json), RedisEvent.class);
	}
}